package com.atlansheng.rocketmqdemoboot.demos.producer;

import com.alibaba.fastjson.JSON;
import com.atlansheng.rocketmqdemoboot.demos.Dto.MsgInfo;
import com.atlansheng.rocketmqdemoboot.demos.conf.MyRocketMqTemplete;
import com.atlansheng.rocketmqdemoboot.demos.po.OneOrder;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDate;
import java.time.ZoneId;
import java.util.Date;

/**
 * @ClassName: ProducerSend
 * @Description:
 * @Date 2024/03/04 11:42
 * @Author songshaoqiu
 **/
@Component
@Slf4j
public class ProducerSend {

    @Autowired
    private MyRocketMqTemplete rocketMQTemplate;

    private final String tagTopic ="tagTopic";
    private final  String topic ="studentTopic";

    private final String failTopic="studentFailTopic";

    private final String transTopic ="transactionTopic";

    /**
     * 单向发送消息。不需要等待Broker的回应
     */
    public void sendOneWay(){
        MsgInfo msgInfo = new MsgInfo();
        msgInfo.setName("张三");
        msgInfo.setAge(11);
        msgInfo.setId(1l);

        Message<String> message = MessageBuilder.withPayload(JSON.toJSONString(msgInfo))
                //设置消息key
                .setHeader("KEYS", msgInfo.getId()).build();
        //发送主题
        String topic ="studentTopic";
        rocketMQTemplate.sendOneWay(topic,message);

    }

    /**
     * 同步发送消息
     */
    public void syncSend(Long id, String name){
        MsgInfo msgInfo = new MsgInfo();
        msgInfo.setId(id);
        msgInfo.setAge(24);
        msgInfo.setName(name);

        Message<String> message = MessageBuilder.withPayload(JSON.toJSONString(msgInfo)).setHeader("KEYS", msgInfo.getId()).build();
        String thisTopic ="";
        if (!id.equals(3L)){
            thisTopic = topic;
        }else{
            thisTopic=failTopic;
        }
        SendResult sendResult = rocketMQTemplate.syncSend(thisTopic, message);
        log.info("消息发送结果{}---->topic-->{}",sendResult,thisTopic);

    }

    public void sendAsync(Long id, String name){
        MsgInfo msgInfo = new MsgInfo();
        msgInfo.setId(id);
        msgInfo.setName(name);
        msgInfo.setAge(66);

        Message<String> message = MessageBuilder.withPayload(JSON.toJSONString(msgInfo))
                .setHeader("KEYS", msgInfo.getId())
                .setHeader(RocketMQHeaders.TAGS,"hello")
                .build();
        rocketMQTemplate.asyncSend(topic, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("发送结果,{}",sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.info("发送失败");
            throwable.printStackTrace();
            }
        });

    }

    /**
     * 延迟消息
     * @param id
     * @param name
     */
    public void sendDelay(Long id, String name){
        MsgInfo msgInfo = new MsgInfo();
        msgInfo.setId(id);
        msgInfo.setName(name);
        msgInfo.setAge(66);

        Message<String> message = MessageBuilder.withPayload(JSON.toJSONString(msgInfo))
                .setHeader("KEYS", msgInfo.getId())
                .build();
        SendResult sendResult = rocketMQTemplate.syncSendDelayTimeMills(topic, message, 5 * 1000);
        log.info("模拟发送5秒后消费消息{}",sendResult);
    }

    /**
     * 定时消息
     * @param id
     * @param name
     */
    public void sendDeliver(Long id, String name){
        MsgInfo msgInfo = new MsgInfo();
        msgInfo.setId(id);
        msgInfo.setName(name);
        msgInfo.setAge(66);

        Message<String> message = MessageBuilder.withPayload(JSON.toJSONString(msgInfo))
                .setHeader("KEYS", msgInfo.getId())
                .build();

        long time = LocalDate.now().atStartOfDay().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
        SendResult sendResult = rocketMQTemplate.syncSendDeliverTimeMills(topic, message, time);

    }

    /**
     * 事务消息
     */
    public void sendTransMsg(Long id, String name){
        MsgInfo msgInfo = new MsgInfo();
        msgInfo.setId(id);
        msgInfo.setName(name);
        msgInfo.setAge(66);
        msgInfo.setType(1);

        Message<String> message = MessageBuilder.withPayload(JSON.toJSONString(msgInfo))
                .setHeader("KEYS", msgInfo.getId())
                .setHeader(RocketMQHeaders.TRANSACTION_ID,"KEY_"+msgInfo.getId())
                .build();
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer();
        TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(topic, message, null);
        log.info("事务消息发送结果{}",transactionSendResult);
    }


    @Transactional
    public void trans(Long id) {
        OneOrder oneOrder = new OneOrder();
        oneOrder.setOrderNo("zfdd"+id);
        oneOrder.setMoney(100);
        oneOrder.setType(1);
        oneOrder.setCreateTime(new Date());

        Message<String> message = MessageBuilder.withPayload(JSON.toJSONString(oneOrder))
                .setHeader("KEYS", oneOrder.getOrderNo())
                .setHeader(RocketMQHeaders.TRANSACTION_ID, "key_" + oneOrder.getOrderNo())
                .build();
        TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(transTopic, message, null);
        log.info("事务消息发送状态{}",transactionSendResult.getLocalTransactionState());
        log.info("完整数据结果{}",transactionSendResult);

    }

    // public void sendWithTag(Long id,String name){
    //     Message<String> message01 = MessageBuilder.withPayload("rocketmq过滤消息测试01").setHeader(RocketMQHeaders.TAGS, "tag01").build();
    //     Message<String> message02 = MessageBuilder.withPayload("rocketmq过滤消息测试02").setHeader(RocketMQHeaders.TAGS, "tag02").build();
    //     Message<String> message03 = MessageBuilder.withPayload("rocketmq过滤消息测试03").setHeader(RocketMQHeaders.TAGS, "tag03").build();
    //
    //     SendResult sendResult = rocketMQTemplate.syncSend(tagTopic, message01);
    //     System.out.println(sendResult);
    //     SendResult sendResult1 = rocketMQTemplate.syncSend(tagTopic, message02);
    //     System.out.println(sendResult1);
    //     SendResult sendResult2 = rocketMQTemplate.syncSend(tagTopic, message03);
    //     System.out.println(sendResult2);
    // }

    /**
     * 发送带Tag消息，测试根据Tag过滤消息
     */
    public void sendWithTag(Long id,String name){
        // 构造消息1
        Message msg1 = MessageBuilder.withPayload("rocketmq过滤消息测试-TAG01").build();
        // 构造消息2
        Message msg2 = MessageBuilder.withPayload("rocketmq过滤消息测试-TAG02").build();
        // 构造消息3
        Message msg3 = MessageBuilder.withPayload("rocketmq过滤消息测试-TAG03").build();

        rocketMQTemplate.convertAndSend("java1234-filter-rocketmq" + ":" + "TAG1", msg1);
        rocketMQTemplate.convertAndSend("java1234-filter-rocketmq" + ":" + "TAG2", msg2);
        rocketMQTemplate.convertAndSend("java1234-filter-rocketmq" + ":" + "TAG3", msg3);

    }

}
